﻿using System.Threading.Channels;

namespace Digitalmes.Infrastructure.MessageBrokers;

/// <summary>
/// 基于内存 <see cref="Channel{}"/> 的消息队列基类。
/// </summary>
/// <typeparam name="T"></typeparam>
public abstract class ChannelQueueBase<T>
{
    private readonly Channel<T> _queue;

    /// <summary>
    /// 初始化一个新的 Broker 对象。
    /// </summary>
    /// <param name="capacity">容量</param>
    /// <param name="mode">通道满载后的处理模式</param>
    /// <param name="isSingleReader">是否只允许单一对象读取</param>
    /// <param name="isSingleWriter">是否只允许单一对象写入</param>
    public ChannelQueueBase(int capacity, BoundedChannelFullMode mode = BoundedChannelFullMode.Wait, bool isSingleReader = true, bool isSingleWriter = false)
    {
        // 多个写入端，单个读取端
        _queue = Channel.CreateBounded<T>(new BoundedChannelOptions(capacity)
        {
            FullMode = mode,
            SingleReader = isSingleReader,
            SingleWriter = isSingleWriter,
        });
    }

    /// <summary>
    /// 将事件写入队列。
    /// </summary>
    /// <param name="item">加入队列的事件项</param>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public async ValueTask EnqueueAsync(T item, CancellationToken cancellationToken = default)
    {
        await _queue.Writer.WriteAsync(item, cancellationToken).ConfigureAwait(false);
    }

    /// <summary>
    /// 异步读取，若队列中没有数据项时会异步阻塞当前线程。
    /// </summary>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public async ValueTask<T> DequeueAsync(CancellationToken cancellationToken = default)
    {
        // 效果等于 WaitToReadAsync + TryRead
        return await _queue.Reader.ReadAsync(cancellationToken).ConfigureAwait(false);
    }
}
